package com.gugee.tiktok.data.consumer.blogger.listener;

import com.alibaba.nacos.api.config.annotation.NacosValue;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.common.msg.HashtagMessage;
import com.gugee.tiktok.data.consumer.blogger.core.main.UpdateBloggerMain;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 *
 * @author luolifeng
 * Date  2020/10/04
 */
@Component
public class BloggerMessageListener {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    UpdateBloggerMain updateBloggerMain;

    @NacosValue(value="${tiktok.data.blogger.accept.minTime.limit:1440}", autoRefreshed = true)
    long acceptMsgTimeMinRange;

    @KafkaListener(id="${kafka.consumer.group.blogger.id}",topics = {"${kafka.topics.blogger.online}"},containerFactory = "factoryBloggerAck")
    public void listen(List<BloggerMessage> messages, Acknowledgment ack) {
        long start=System.currentTimeMillis();
        long acceptMsgTimeSecRange=acceptMsgTimeMinRange*60*1000;
        String error="success";
        try {
            logger.info("接收到消息数量: {}" , messages.size());
            long isRange=start-acceptMsgTimeSecRange;
            /**
             * 消息如果不在处理时限内，则丢弃
             */
            if(messages.get(0).getEventTime()>isRange) {
                updateBloggerMain.handle(messages);
            }else{
                logger.info("消息数量: {} ,不在处理时限内，跳过;第一消息时间：{},时限分钟：{}" , messages.size(),messages.get(0).getEventTime(),acceptMsgTimeMinRange );
            }
            ack.acknowledge();
        }catch (Exception e){
//            e.printStackTrace();
            logger.info("发生异常: {}" , e.getMessage());
            error=e.getMessage();
        }finally {
            long times=(System.currentTimeMillis()-start)/1000;
            if(times<1){
                times=1;
            }
            logger.info("消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,结果：{}" , messages.size(),times,messages.size()/times,error);
        }
    }

//    @KafkaListener(id="${kafka.consumer.group.hashtagGroup.id}",topics = {"${kafka.topics.hashtag.online}"},containerFactory = "factoryHashtagAck")
//    public void listenHashtag(List<HashtagMessage> messages, Acknowledgment ack) {
//        long start=System.currentTimeMillis();
//        try {
//            logger.info("接收到消息数量: {},第一个消息：{}" , messages.size(),messages.get(0));
//            //todo 开始你的表演
//        }catch (Exception e){
//            e.printStackTrace();
//        }finally {
//            long ackTime=System.currentTimeMillis();
//            ack.acknowledge();
//            long times=(System.currentTimeMillis()-start)/1000;
//            if(times<1){
//                times=1;
//            }
//            logger.info("消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,确认消息耗时：{}" , messages.size(),times,messages.size()/times,(System.currentTimeMillis()-ackTime)/1000);
//        }
//    }
}
